//package com.world.rickcloudendpoint.service.runner;
//
//import com.world.rickcloudendpoint.config.ConsumerConfig;
//import com.world.rickcloudendpoint.stream.consumer.RickCloudConsumer;
//import com.world.rickcloudendpoint.stream.consumer.RickCloudConsumerFactory;
//import com.world.rickcloudendpoint.metrics.Instrumentation;
//import lombok.extern.log4j.Log4j2;
//import org.springframework.boot.CommandLineRunner;
//import org.springframework.core.annotation.Order;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//
///**
// * @description:
// * @author: jlsong
// * @date: 2022/02/28 22:52:21
// * @version: 1.0
// */
//@Component
//@Order(10)
//@Log4j2
//public class MainRunner implements CommandLineRunner {
//
//    final ConsumerConfig consumerConfig;
//
//    public MainRunner(ConsumerConfig consumerConfig) {
//        this.consumerConfig = consumerConfig;
//    }
//
//    @Override
//    public void run(String... args) throws Exception {
//        // 多线程消费者
//        multiThreadedConsumers(consumerConfig);
//
//    }
//
//    /**
//     * 多线程消费
//     *
//     * @param config 消费者配置i
//     */
//    private void multiThreadedConsumers(ConsumerConfig config) throws InterruptedException {
//        // 配置里面获取数目
//        log.info("Number of consumer threads: {} ", config.getApplicationThreadCount());
//        log.info("Delay to clean up consumer threads in ms: {}", config.getApplicationThreadCleanupDelay());
//
//        Task consumerTask = new Task(
//                config.getApplicationThreadCount(),
//                config.getApplicationThreadCleanupDelay(),
//                new Instrumentation(),
//                taskFinished -> {
//                    RickCloudConsumer rickCloudConsumer = null;
//                    try {
//                        rickCloudConsumer = new RickCloudConsumerFactory(consumerConfig).buildConsumer();
//                        while (true) {
//                            if (Thread.interrupted()) {
//                                log.warn("Consumer Thread interrupted,leaving the loop!");
//                                break;
//                            }
//                            rickCloudConsumer.process();
//                        }
//                    } catch (Exception e) {
//                        log.info("Exception on creating the consumer, exiting the application");
//                        System.exit(1);
//                    } finally {
//                        ensureThreadInterruptStateIsClearedAndClose(rickCloudConsumer);
//                        taskFinished.run();
//                    }
//                }
//        );
//        log.info("Consumer task created");
//        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
//            log.info("Program is going to exit.Have started execution of shutdownHook before this");
//            consumerTask.stop();
//        }));
//
//        consumerTask.run().waitForCompletion();
//        ;
//        log.info("Exiting main thread");
//    }
//
//    private void ensureThreadInterruptStateIsClearedAndClose(RickCloudConsumer consumer) {
//        Thread.interrupted();
//        try {
//            consumer.close();
//        } catch (IOException e) {
//            log.info("Closing consumer failed,the reason is {} ", e.getMessage(), e);
//        }
//    }
//}
